package org.caw.hbase;

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.LocalHBaseCluster;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;

public class MiniLocalHBase {

	private static final String ROOT = "mini_hbase";
	private static final String ZK_PATH = ROOT + "/zk";
	private static final String HBASE = ROOT + "/hbase";

	static LocalHBaseCluster hbaseCluster;
	static MiniZooKeeperCluster zkCluster;

	public static void main(String[] args) throws IOException, InterruptedException {

		deleteFolder(new File(ROOT));
		zkCluster = new MiniZooKeeperCluster();
		zkCluster.setClientPort(2181);

		Thread.sleep(2000);

		Configuration conf = HBaseConfiguration.create();
		conf.set(HConstants.HBASE_DIR, HBASE);
		conf.set("hbase.regionserver.info.port", "-1");
		zkCluster.startup(new File(ZK_PATH));
		hbaseCluster = new LocalHBaseCluster(conf, 1);
		hbaseCluster.startup();

		Thread.sleep(2000);
		createSampleSchema(conf);
	}

	public static void deleteFolder(File f) throws IOException {
		if (f.exists()) {
			if (f.isDirectory()) {
				for (File c : f.listFiles())
					deleteFolder(c);
			}
			if (!f.delete()) {
				throw new IOException("Failed to delete file: " + f);
			}
		}
	}

	public static void createSampleSchema(Configuration conf) throws IOException {
		HBaseAdmin admin = new HBaseAdmin(conf);

		HTableDescriptor table1 = new HTableDescriptor("TABLE_1");
		table1.addFamily(new HColumnDescriptor("CF1"));
		table1.addFamily(new HColumnDescriptor("CF2"));
		admin.createTable(table1);

		HTableDescriptor table2 = new HTableDescriptor("TABLE_2");
		table2.addFamily(new HColumnDescriptor("FC1"));
		admin.createTable(table2);

		HTable t1 = new HTable(conf, "TABLE_1");
		Put p1 = new Put(Bytes.toBytes("1"));
		p1.add(Bytes.toBytes("CF1"), Bytes.toBytes("q1"), Bytes.toBytes("value1"));
		p1.add(Bytes.toBytes("CF1"), Bytes.toBytes("q2"), Bytes.toBytes("value2"));
		p1.add(Bytes.toBytes("CF1"), Bytes.toBytes("q3"), Bytes.toBytes("value3"));
		p1.add(Bytes.toBytes("CF2"), Bytes.toBytes("q2"), Bytes.toBytes("value1"));

		t1.put(p1);

		Put p2 = new Put(Bytes.toBytes("2"));
		p2.add(Bytes.toBytes("CF2"), Bytes.toBytes("q1"), Bytes.toBytes("value1"));
		p2.add(Bytes.toBytes("CF2"), Bytes.toBytes("q2"), Bytes.toBytes("value2"));
		p2.add(Bytes.toBytes("CF2"), Bytes.toBytes("q3"), Bytes.toBytes("value3"));
		p2.add(Bytes.toBytes("CF2"), Bytes.toBytes("q4"), Bytes.toBytes("value4"));

		t1.put(p2);

		t1.close();

	}

}